package com.navinfo.opentsp.platform.computing.analysis.util;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class AreaMap implements Serializable {

    private static volatile Broadcast<HashMap<Integer, String>> instance = null;

    public static Broadcast<HashMap<Integer, String>> getInstance(SparkSession spark, JavaSparkContext jsc, String sql) {
        if (instance == null) {
            synchronized (TileMap.class) {
                if (instance == null) {
                    instance = jsc.broadcast(loadTileMap(spark,sql));
                }
            }
        }
        return instance;
    }

    private static HashMap<Integer,String> loadTileMap (SparkSession spark,String sql){

        //广播经纬度对照省市区信息
        Dataset<Row> areaSet = spark.sql(sql);
        JavaRDD<Row> areaRdd = areaSet.toJavaRDD();
        JavaPairRDD<Integer,String> areaPairRdd =
                areaRdd.mapPartitionsToPair(
                        (PairFlatMapFunction<Iterator<Row>, Integer, String>) rows ->
                                new Iterator<Tuple2<Integer, String>>() {
                                    @Override
                                    public boolean hasNext() {
                                        return rows.hasNext();
                                    }

                                    @Override
                                    public Tuple2<Integer, String> next() {
                                            Row row = rows.next();
                                            int ccode = Integer.parseInt(row.getAs("city_code"));
                                            String cname = row.getAs("city_name");
                                            int pcode = Integer.parseInt(row.getAs("province_code"));
                                            String pname = row.getAs("province_name");
                                            //如果是直辖市或特别行政区，则将城市代码替换成省市代码，与瓦片表保持一致
                                            //直辖市只取第一个城市编号修改（针对重庆市，有两个市级编号）
                                            boolean flag = ccode - pcode == 100;
                                            if(pname.equalsIgnoreCase(cname) && flag){
                                                ccode = pcode;
                                            }
                                            StringBuffer sb = new StringBuffer();
                                            String value = sb.append(ccode).append("|").append(cname)
                                                    .append(",").append(pcode).append("|").append(pname).toString();
                                        return new Tuple2<>(ccode,value);
                                    }
                                });
        Map<Integer,String> map = areaPairRdd.collectAsMap();
        HashMap<Integer,String> areaMap = new HashMap<>(map);
        return areaMap;
    }
}
